package net.bw.realtime.tmall.dws.job;

import com.alibaba.fastjson.JSON;
import net.bw.realtime.tmall.common.utils.JdbcUtil;
import net.bw.realtime.tmall.common.utils.KafkaUtil;
import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import net.bw.realtime.tmall.dws.function.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-27
 */
public class TmallStatHourWindowDwsJob {

    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2.  读取数据
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "transaction_event_bean-topic-abnormal");

        //kafkaStream.print();

        // 3. 数据转换
        SingleOutputStreamOperator<TransactionEventBean> beanStream = kafkaStream.process(
                new ProcessFunction<String, TransactionEventBean>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, TransactionEventBean>.Context ctx, Collector<TransactionEventBean> out) throws Exception {
                        TransactionEventBean eventBean = JSON.parseObject(value, TransactionEventBean.class);
                        out.collect(eventBean);
                    }
                }
        );

        // 4.设置水位线
        SingleOutputStreamOperator<TransactionEventBean> waterMarksStream = beanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TransactionEventBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TransactionEventBean>() {
                                    @Override
                                    public long extractTimestamp(TransactionEventBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        //waterMarksStream.print();

        //  4.实时交易额统计
//        toTotalAmount(waterMarksStream);

        //  5.实时订单量统计
        toOrderCount(waterMarksStream);

        //  6.热门商品排行
//        toSkuTop(waterMarksStream);

        //  7.地区分布统计
//        toAreaDistribution(waterMarksStream);

        //  8.异常交易监控
//        toCepTrade(waterMarksStream);

        //  9.设备分布统计
        toDeviceDistribution(waterMarksStream);

        env.execute("TmallOrderStatHourWindowDwsJob");

    }

    // 9.设备分布统计
    private static void toDeviceDistribution(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        SingleOutputStreamOperator<String> deviceStream = waterMarksStream
                .filter(x -> "pay_success".equals(x.getActionType()))
                .keyBy(bean -> bean.getDeviceType())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .apply(new TmallDeviceDistributionFunction());

//        deviceStream.print();

//        JdbcUtil.sinkToClickhouseUpsert(
//                deviceStream,
//                "INSERT INTO jtp_tmall_report.dws_device_distribution_report(\n" +
//                        "    window_start_time, window_end_time, device_type, total_amount, ts\n" +
//                        ") VALUES (\n" +
//                        "    ?, ?, ?, ?, ?\n" +
//                        ")"
//        );

    }

    // 8.异常交易监控
    private static void toCepTrade(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        // 定义异常模式，同一用户在短时间内多次大额支付
        Pattern<TransactionEventBean, ?> pattern = Pattern.<TransactionEventBean>begin("first").where(
                new SimpleCondition<TransactionEventBean>() {
                    @Override
                    public boolean filter(TransactionEventBean TransactionEventBean) throws Exception {

                        return "pay_success".equals(TransactionEventBean.getActionType())
                                && (TransactionEventBean.getPrice() * TransactionEventBean.getQuantity()) > 10000;
                    }

                }
        ).next("second").where(
                new SimpleCondition<TransactionEventBean>() {
                    @Override
                    public boolean filter(TransactionEventBean TransactionEventBean) throws Exception {
                        return "pay_success".equals(TransactionEventBean.getActionType())
                                && (TransactionEventBean.getPrice() * TransactionEventBean.getQuantity()) > 10000;
                    }
                }
        ).within(Time.minutes(1));

        KeyedStream<TransactionEventBean, String> keyedStream = waterMarksStream.keyBy(TransactionEventBean::getUserId);

        SingleOutputStreamOperator<String> cep_trade = CEP.pattern(keyedStream, pattern)
                .select(new PatternSelectFunction<TransactionEventBean, String>() {
                    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    @Override
                    public String select(Map<String, List<TransactionEventBean>> pattern) throws Exception {
                        TransactionEventBean first = pattern.get("first").get(0);
                        TransactionEventBean second = pattern.get("second").get(0);

                        String result = "用户：" + first.getUserId() + "在" + first.getTs() + "发生异常交易，金额：" + first.getPrice() * first.getQuantity() + "元" +
                                "，在" + second.getTs() + "又发生异常交易，金额：" + second.getPrice() * second.getQuantity() + "元";

						//System.out.println(result);

                        return first.getUserId() + "," + sdf.format(new Date(first.getTs())) + "," + first.getPrice() * first.getQuantity() + ","  + sdf.format(new Date(second.getTs())) + "," + second.getPrice() * second.getQuantity() ;
                    }
                })
                .name("cep-trade-choose")
                .uid("cep-trade-choose");

		cep_trade.print();


    }

    // 7.地区分布统计
    private static void toAreaDistribution(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        SingleOutputStreamOperator<String> areaStream = waterMarksStream
                .filter(x -> "pay_success".equals(x.getActionType()))
                .keyBy(bean -> bean.getProvince() + "," + bean.getCity())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .apply(new TmallAreaDistributionFunction());

        //areaStream.print();

        JdbcUtil.sinkToClickhouseUpsert(
                areaStream,
                "INSERT INTO jtp_tmall_report.dws_area_daily_report (\n" +
                        "    window_start_time, window_end_time, province, city, order_count, total_amount, ts\n" +
                        ") VALUES (\n" +
                        "    ?, ?, ?, ?, ?, ?, ?\n" +
                        ")"
        );

    }

    // 6.热门商品排行
    private static void toSkuTop(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        SingleOutputStreamOperator<String> skuCount = waterMarksStream
                .filter(x -> "pay_success".equals(x.getActionType()))
                .keyBy(bean -> bean.getSkuId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .apply(new TmallSkuTopFunction());

        skuCount.print();

//        JdbcUtil.sinkToClickhouseUpsert(
//                skuCount,
//                "INSERT INTO jtp_tmall_report.dws_sku_count_window_report (\n" +
//                        "     window_start_time, window_end_time, sku_name, sku_count, sku_total_amount, ts\n" +
//                        ") VALUES (\n" +
//                        "    ?, ?, ?, ?, ?\n" +
//                        ")"
//        );

    }

    // 5.实时订单量统计
    private static void toOrderCount(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        waterMarksStream
                .filter(x -> "place_order".equals(x.getActionType()))
                .keyBy(x -> "sku_id")
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .aggregate(new AggregateFunction<TransactionEventBean, Long, Long>() {

                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(TransactionEventBean tmallLogBean, Long aLong) {
                        return aLong + 1;
                    }

                    @Override
                    public Long getResult(Long aLong) {

                        return aLong;
                    }

                    @Override
                    public Long merge(Long aLong, Long acc1) {
                        return aLong + acc1;
                    }
                }, new ProcessWindowFunction<Long, String, String, TimeWindow>() {
                    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    @Override
                    public void process(String s, ProcessWindowFunction<Long, String, String, TimeWindow>.Context context, Iterable<Long> iterable, Collector<String> collector) throws Exception {
                        Long next = iterable.iterator().next();
                        String window_start = sdf.format(new Date(context.window().getStart()));
                        String window_end = sdf.format(new Date(context.window().getEnd()));

                        collector.collect(window_start + "," + window_end + "," + next + "," + System.currentTimeMillis());
                    }
                })
                .name("order-count-aggregation")
                .uid("order-count-aggregation")
                .print();


        // 计算订单总数
        SingleOutputStreamOperator<String> orderCount = waterMarksStream
                .filter(x -> "pay_success".equals(x.getActionType()))
                .keyBy(e -> "orderCount")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .apply(new TmallOrderCountFunction());

        //orderCount.print();

//        JdbcUtil.sinkToClickhouseUpsert(
//                orderCount,
//                "INSERT INTO jtp_tmall_report.dws_order_count_window_report (\n" +
//                        "     window_start_time, window_end_time, order_count, ts\n" +
//                        ") VALUES (\n" +
//                        "    ?, ?, ?, ?\n" +
//                        ")"
//        );
    }

    // 4.实时交易额统计
    private static void toTotalAmount(SingleOutputStreamOperator<TransactionEventBean> waterMarksStream) {

        // 计算交易总额
        SingleOutputStreamOperator<String> totalAmount = waterMarksStream
                .filter(x -> "pay_success".equals(x.getActionType()))
                .keyBy(e -> "totalAmount")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .apply(new TmallTotalAmountFunction());

        //totalAmount.print();

        JdbcUtil.sinkToClickhouseUpsert(
                totalAmount,
        "INSERT INTO jtp_tmall_report.dws_order_total_amount_window_report (\n" +
                "     window_start_time, window_end_time, total_amount, ts\n" +
                ") VALUES (\n" +
                "    ?, ?, ?, ?\n" +
                ")"
        );

    }

}
